package net.xt.testthread.utils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

@Slf4j
public class MoreThreadTransactionUtils {

    /**
     * 多线程事务处理
     * 多线程事务批量插入，故意出bug，演示要么一起成功，要么全体连坐
     *
     * @param platformTransactionManager Spring的事务管理器，用于控制事务的提交和回滚
     * @param taskList                   一个可变数量的Runnable任务，每个任务代表一个数据库操作
     * @return 如果所有任务都成功完成，则返回true；否则返回false
     */
    public static boolean execute(PlatformTransactionManager platformTransactionManager, Runnable... taskList) {
        if (taskList == null || taskList.length == 0) {
            throw new IllegalArgumentException("taskList is empty");
        }

        // 任务数量
        int taskSize = taskList.length;
        // 任务成功数量计数器
        AtomicInteger taskSuccessAccount = new AtomicInteger(0);

        List<Future<?>> taskFutureList = new ArrayList<>(taskSize);

        // 循环屏障，用于让多线程事务要么一起成功，要么全体连坐
        CyclicBarrier cyclicBarrier = new CyclicBarrier(taskSize);
        int i = 1;
        // 定义了一个线程池，线程池核心线程数大小和任务大小必须一致，因为里面的任务都必须同时去执行，否则会死锁
        ExecutorService threadPool = Executors.newFixedThreadPool(taskSize);

        try {
            //使用线程池执行循环处理任务，每个任务会交给线程池中的一个线程执行
            for (Runnable runnableThread : taskList) {
                final int taskIndex = i;

                Future<?> future = threadPool.submit(() -> {
                    TransactionStatus transactionStatus = null;
                    try {
                        // 使用spring编程式事务，开启事务
                        transactionStatus = platformTransactionManager.getTransaction(new DefaultTransactionAttribute());
                        // 执行任务
                        runnableThread.run();
                        // 成功数量+1
                        taskSuccessAccount.incrementAndGet();
                        log.info("task：{} 等待提交事务", taskIndex);
                    } catch (Throwable e) {
                        log.error("task：{}，执行异常，异常原因：{}", taskIndex, e.getMessage());
                    } finally {
                        try {
                            //运行到这，cyclicBarrier阻塞发生，当池中所有任务都over才会被唤醒，继续向下走
                            cyclicBarrier.await();
                        } catch (Exception e) {
                            log.error("cyclicBarrier.await error:{}", e.getMessage(), e);
                        }
                    }

                    //根据事务返回的状态确认commit or  rollback
                    if (transactionStatus != null) {
                        // 如果所有任务都成功（successAccount的值等于任务总数），则一起提交事务，如果有任何任务失败，则一起回滚事务
                        if (taskSuccessAccount.get() == taskSize) {
                            // 成功，提交事务
                            log.info("task：{} 提交事务", taskIndex);
                            platformTransactionManager.commit(transactionStatus);
                        } else {
                            //失败，回滚事务
                            log.info("task：{} 回滚事务", taskIndex);
                            platformTransactionManager.rollback(transactionStatus);

                            //todo
                            //出错消息发送到钉钉或者告警短信或者其它专属异常记录数据库，用xxl-job再
                            //定时重试，进入下一次排期
                        }
                    }
                });

                taskFutureList.add(future);
                i++;
            }

            for (Future<?> future : taskFutureList) {
                try {
                    future.get();
                } catch (Exception e) {
                    log.error("future.get error:{}", e.getMessage(), e);
                }
            }
        } finally {
            //关闭线程池
            threadPool.shutdown();
        }

        //如果所有任务都成功完成，则返回true；否则返回false
        return taskSuccessAccount.get() == taskSize;
    }

    public static boolean execute2(PlatformTransactionManager platformTransactionManager,
                                   Supplier... taskList) {
        // 任务成功数量计数器
        AtomicInteger sum = new AtomicInteger(0);
        // 循环屏障，用于让多线程事务要么全部成功提交，要么全部失败回滚
        CyclicBarrier cyclicBarrier = new CyclicBarrier(taskList.length);
        // 定义一个自定义线程池，固定大小为任务的数量
        ExecutorService threadPool = Executors.newFixedThreadPool(taskList.length);
        List<CompletableFuture> list = new ArrayList<>();

        for (Supplier supplier : taskList) {
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                // 获取当前事务的状态
                TransactionStatus transactionStatus = platformTransactionManager.getTransaction(new DefaultTransactionAttribute());

                supplier.get(); // 执行事务
                sum.incrementAndGet();  // 成功数+1

                try {
                    //运行到这，cyclicBarrier阻塞当前线程，当所有线程执行完任务才会被唤醒，继续向下走
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("cyclicBarrier.await error:{}", e.getMessage(), e);
                }

                // 事务提交或回滚
                if (transactionStatus != null) {
                    if (sum.get() == taskList.length) {
                        platformTransactionManager.commit(transactionStatus);
                        log.info("task：{} 提交事务", Thread.currentThread().getName());
                    } else {
                        platformTransactionManager.rollback(transactionStatus);
                        log.info("task：{} 回滚事务", Thread.currentThread().getName());
                    }
                }
            }, threadPool);

            list.add(future);
        }

        // 多线程执行
        list.forEach(completableFuture -> completableFuture.join());
        // 线程池关闭
        threadPool.shutdown();

        return sum.get() == taskList.length;
    }

}
